package com.catmiao.spark.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: RDD_Operator_01_transform
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/2/2 11:48
 */
object RDD_Operator_16_transform {

  def main(args: Array[String]): Unit = {
    val sparkCon = new SparkConf().setMaster("local[*]").setAppName("rdd")

    val sparkContext = new SparkContext(sparkCon)


    val rdd = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)),2)

    // 分区内求最大值 分区间求和

    /**
     * aggregateByKey 存在函数柯里化
     *      param1: 传递一个参数，表示初始值，主要为了确定遇到第一个值时进行分区内的聚合计算
     *      param2: 传递两个函数
     *          param1：分区内的计算规则
     *          param2：分区间的计算规则
     */
    val rdd2: RDD[(String, Int)] = rdd.aggregateByKey(0)(
      (x, y) => math.max(x, y),
      (x, y) => x + y
    )

    // 如果分区规则相同，提供简化的方法
    // rdd.foldByKey(0)(_+_)

    rdd2.collect().foreach(println)

    sparkContext.stop()
  }

}
